package com.alks.pad.controller.last;

import com.alibaba.fastjson.JSON;
import com.alks.common.service.RedisService;
import com.alks.common.utils.IpAddressUtils;
import com.alks.entity.data.response.ResponseInfo;
import com.alks.entity.exception.EmptyDataException;
import com.alks.pad.commom.date.thread.PadUserIdThread;
import com.alks.pad.service.last.LastCollectDetailService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.servlet.http.HttpServletRequest;
import java.net.ConnectException;
import java.time.Duration;

@Slf4j
@RestController
@RequestMapping("/pad/last")
@Api(tags = {"sse连接"})
public class SseController {

    @Autowired
    private LastCollectDetailService lastCollectDetailService;
    @Autowired
    private RedisService redisService;

    private final String SSE_CLIENT_KEY = "SSE:%s:%s";

    @GetMapping(value = "/sizeByLot", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @ApiOperation(value = "查询号码明细",tags = "sse连接")
    public Flux<ResponseInfo> streamSizeByLotNo(@RequestParam String lotNo, HttpServletRequest request) {
        String companyId = PadUserIdThread.get().getCompanyId();
        String host = IpAddressUtils.getClientIpAddress(request);
        String port = String.valueOf(request.getRemotePort());
        String key = String.format(SSE_CLIENT_KEY, host, PadUserIdThread.get().getCollectUserName());
        redisService.set(key, port, 8 * 60 * 60);
        // 定时每 5 秒调用一次 SizeByLotNo 方法，返回数据流
        return Flux.interval(Duration.ofSeconds(5))
                .flatMap(sequence -> {
                    ResponseInfo responseInfo = lastCollectDetailService.SizeByLotNo(companyId, lotNo);
                    if (!port.equals(redisService.get(key))) {
                        log.info("客户端{}连接重置 {} -> {}", key, port, redisService.get(key));
                        return Flux.error(new ConnectException("客户端断开连接"));
                    }
                    return Mono.fromCallable(() -> responseInfo);
                })
                .onErrorResume(e -> {
                    // 捕获异常，记录日志或返回自定义响应
                    log.info("连接已关闭: " + e.getMessage());
                    e.printStackTrace();
                    return Flux.error(new ConnectException("连接已关闭"));
                });
    }


}